package com.chukun.flink.stream.kafka;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;

/**
 * @author chukun
 * @version 1.0.0
 * @description 实现kafka的分区器
 * @createTime 2022年06月04日 12:05:00
 */
public class KafkaMessagePartition extends FlinkKafkaPartitioner<Tuple2<Integer,String>> {
    @Override
    public int partition(Tuple2<Integer, String> message, byte[] key, byte[] value, String topic, int[] partitions) {
        System.out.println("数据流中的元素为: "+message.toString());
        System.out.println("消息key值: "+new String(key));
        System.out.println("消息: "+new String(value));
        System.out.println("topic: "+topic);

        // 处理消息分区
        Integer f0 = message.f0;
        return partitions[f0 % partitions.length];
    }
}
